use futures::channel::mpsc;
use hashbrown::HashMap;
use honeybee::prelude::*;
use honeybee::Result;

use super::user;

struct UserSession {
    session_key: String,
    user_addr: Addr<super::User>,
    user_stream: Addr<super::UserStream>,
}

#[derive(Default)]
pub struct UserCenter {
    sequence_id: u128,
    users: HashMap<i64, UserSession>,
}

impl Service for UserCenter {}

#[async_trait::async_trait]
impl Actor for UserCenter {
    fn name(&self) -> &str {
        "UserCenter"
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        info!("{} started", self.name());

        let router = ctx.get_router().await?;
        router.register::<proto::UserMessageForward>();
        router.register::<proto::AllUserMessageForward>();
        router.register::<proto::UserMessageApp>();
        router.register::<proto::AllUserMessageApp>();

        Ok(())
    }

    async fn stopped(&mut self, _ctx: &mut Context<Self>) {
        info!("{} stoped", self.name());
    }
}

#[message(result = "u128")]
pub struct Register {
    pub uid: i64,
    pub session_key: String,
    pub user_addr: Addr<super::User>,
    pub user_stream: Addr<super::UserStream>,
}

#[async_trait::async_trait]
impl Handler<Register> for UserCenter {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: Register) -> Result<u128> {
        let user_session = UserSession {
            session_key: msg.session_key,
            user_addr: msg.user_addr,
            user_stream: msg.user_stream,
        };
        let old_user = self.users.insert(msg.uid, user_session);
        if let Some(old_user) = old_user {
            old_user
                .user_addr
                .send(user::Kickout("user_center repeat online".into()))
                .ok();
        }
        self.sequence_id = self.sequence_id.wrapping_add(1);
        Ok(self.sequence_id)
    }
}

#[message]
pub struct Unregister {
    pub uid: i64,
    pub session_key: String,
}

#[async_trait::async_trait]
impl Handler<Unregister> for UserCenter {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: Unregister) -> Result<()> {
        if let Some(user) = self.users.get(&msg.uid) {
            if user.session_key != msg.session_key {
                return Ok(());
            }
        }
        if self.users.remove(&msg.uid).is_none() {
            return Err(anyhow!(
                "user center unregister:{}, not found user",
                msg.uid
            ));
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<NetMsg<proto::UserMessageForward>> for UserCenter {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: NetMsg<proto::UserMessageForward>,
    ) -> Result<()> {
        let msg = msg.msg;
        if let Some(user) = self.users.get(&msg.uid) {
            if let Err(err) = user
                .user_stream
                .send(super::UserMsgForward(msg.msg_id, msg.bytes))
            {
                error!(
                    "forward msg:{} to user:{} failed:{}",
                    msg.msg_id, msg.uid, err
                );
            }
        } else {
            error!(
                "forward msg:{} to user:{} failed:not found user",
                msg.msg_id, msg.uid
            );
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<NetMsg<proto::AllUserMessageForward>> for UserCenter {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: NetMsg<proto::AllUserMessageForward>,
    ) -> Result<()> {
        let msg = msg.msg;
        for (uid, user) in self.users.iter() {
            if let Err(err) = user
                .user_stream
                .send(super::UserMsgForward(msg.msg_id, msg.bytes.clone()))
            {
                error!("forward msg:{} to user:{} failed:{}", msg.msg_id, uid, err);
            }
        }
        Ok(())
    }
}

pub trait RawMsgAddrExt {
    fn send_raw_msg(&self, msg_id: u16, bytes: Vec<u8>) -> Result<()>;
}

impl RawMsgAddrExt for Addr<super::User> {
    fn send_raw_msg(&self, msg_id: u16, bytes: Vec<u8>) -> Result<()> {
        mpsc::UnboundedSender::clone(&*self.tx)
            .start_send(ActorEvent::Exec(Box::new(move |actor, ctx| {
                Box::pin(async move {
                    if let Err(err) =
                        super::appuser_message_process(msg_id, bytes, actor, ctx).await
                    {
                        error!("actor:{} send_raw_msg:{} err:{}", actor.name(), msg_id, err);
                    }
                })
            })))
            .map_err(|err| err.into())
    }
}

#[async_trait::async_trait]
impl Handler<NetMsg<proto::UserMessageApp>> for UserCenter {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: NetMsg<proto::UserMessageApp>,
    ) -> Result<()> {
        let msg = msg.msg;
        if let Some(user) = self.users.get(&msg.uid) {
            if let Err(err) = user.user_addr.send_raw_msg(msg.msg_id, msg.bytes) {
                error!("msg:{} to app user:{} failed:{}", msg.msg_id, msg.uid, err);
            }
        } else {
            error!(
                "msg:{} to app user:{} failed:not found user",
                msg.msg_id, msg.uid
            );
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<NetMsg<proto::AllUserMessageApp>> for UserCenter {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: NetMsg<proto::AllUserMessageApp>,
    ) -> Result<()> {
        let msg = msg.msg;
        for (uid, user) in self.users.iter() {
            if let Err(err) = user.user_addr.send_raw_msg(msg.msg_id, msg.bytes.clone()) {
                error!("msg:{} to app user:{} failed:{}", msg.msg_id, uid, err);
            }
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<proto::NewGmsgNotify> for UserCenter {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: proto::NewGmsgNotify) -> Result<()> {
        for (uid, user) in self.users.iter() {
            if let Err(err) = user.user_addr.send(msg.clone()) {
                error!(
                    "gmsg objectId:{} msg_type:{:?} to app user:{} failed:{}",
                    msg.object_id, msg.msg_type, uid, err
                );
            }
        }
        Ok(())
    }
}
